package sparksql.say8.lesson01

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, StringType, StructField, StructType}

/**
  * Created by Administrator on 2018/5/4.
  *
  * 思路：
  * 声明一个空字符串
  *        字符串的拼接
  */
object GroupDinstictUDAF extends  UserDefinedAggregateFunction{
  /**
    * 输入的数据类型
    * @return
    */
  override def inputSchema: StructType = StructType(
    StructField("city_info",StringType,true) :: Nil
  )

  /**
    * 定义输出的数据类型
    * @return
    */
  override def dataType: DataType = StringType

  /**
    * 定义缓存字段和缓存字段的数据类型
    * @return
    */
  override def bufferSchema: StructType = {
   StructType(
     Array(
       StructField("buffer_city_info",StringType,true)))
  }

  /**
    * 对辅助字段进行初始化
    * @param buffer
    */
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer.update(0,"")
  }

  /**
    * 修改 辅助字段的值
    * @param buffer 上一次的结果
    * @param input 当前输入
    */
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    var last_str = buffer.getString(0)
    val current_str = input.getString(0)
    if(!last_str.contains(current_str)){
      if(last_str.equals("")){//判断是否是第一次
        last_str=current_str
      }else{
        last_str+=","+current_str
      }
    }
    buffer.update(0,last_str)
  }

  /**
    *
    * @param buffer1  hadoop1 的 历史结果
    * @param buffer2  hadoop2 的 历史结果
    */
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit ={
    var b1 = buffer1.getString(0)
    val b2 = buffer2.getString(0)
    for(s <- b2.split(",")){
      if(!b1.contains(s)){
        if(b1.equals("")){
          b1=s
        }else{
          b1+=","+s
        }
      }
    }
    buffer1.update(0,b1)

  }

  /**
    * 最终的计算结果
    * @param buffer
    * @return
    */
  override def evaluate(buffer: Row): Any = {
    buffer.getString(0)
  }
  override def deterministic: Boolean = true
}
